import os, unittest

import aphw

class CtorTests(unittest.TestCase):
    
    def cleanup(self):
        try:
            os.remove(self.db_name)
        except OSError:
            pass

    def setUp(self):
        self.db_name = 'test.db'

        self.cleanup()

    def tearDown(self):
        self.cleanup()

    def test_newFile(self):
        '''Create new Database'''
        self.assert_(not os.path.exists(self.db_name))
        with aphw.Database(self.db_name) as d:
            pass
        self.assert_(os.path.exists(self.db_name))

    def test_existingFile(self):
        '''Open existing Database'''
        self.assert_(not os.path.exists(self.db_name))
        with aphw.Database(self.db_name) as d:
            d.insert(42, 'llama!')
        self.assert_(os.path.exists(self.db_name))
        with aphw.Database(self.db_name) as d:
            self.assert_(str(d.find(42)) == 'llama!')

class Tests(unittest.TestCase):

    def setUp(self):
        self.db_name = 'test.db'

        try:
            os.remove(self.db_name)
        except OSError:
            pass

        self.db = aphw.Database(self.db_name)

    def tearDown(self):
        self.db.close()

        try:
            os.remove(self.db_name)
        except OSError:
            pass

    def test_close(self):
        '''Closing database'''
        self.db.key_count()
        self.db.close()
        try:
            self.db.key_count()
            self.assert_(False)
        except aphw.StatusError, e:
            self.assert_(e.err == aphw.StatusCodes.NOT_INITIALIZED)

    def test_insert(self):
        '''Basic Database insert'''
        for k in [100, aphw.Key(123)]:
            self.db.insert(k, 'jones')
            self.assert_(str(self.db.find(k)) == 'jones')

    def test_insert_duplicate(self):
        '''Database insert fails on duplicates'''
        for k in [100, aphw.Key(123)]:
            self.db.insert(k, 'jones')
            self.assertRaises(KeyError,
                              self.db.insert,
                              k, 'llama')

    def test_insert_overwrite(self):
        '''Database insert overwrite'''
        for k in [100, aphw.Key(321)]:
            self.db.insert(k, 'jones')
            self.db.insert(k, 'llama', overwrite=True)
            self.assert_(str(self.db.find(k)) == 'llama')

    def test_find(self):
        '''Basic Database find'''
        for i in [1, 4321, 9727934, aphw.Key(234), aphw.Key(9898), aphw.Key(4269)]:
            self.db.insert(i, str(i))
            self.assert_(str(self.db.find(i)) == str(i))

    def test_find_unfound(self):
        '''Database find for non-existent keys'''
        for i in [1, 4321, 5418, 3452345, aphw.Key(234), aphw.Key(9898), aphw.Key(4269)]:
            self.assertRaises(KeyError,
                              self.db.find,
                              i)

    def test_erase(self):
        '''Basic Database erase'''
        for k in [123, aphw.Key(12345)]:
            self.db.insert(k, '456')
            self.assert_(str(self.db.find(k)) == '456')
            self.db.erase(k)
            self.assertRaises(KeyError,
                              self.db.find,
                              k)

    def test_erase_unfound(self):
        '''Database erase for non-existent key'''
        for k in [123, aphw.Key(4321)]:
            self.assertRaises(KeyError,
                              self.db.erase,
                              k)
        
    def test_flush(self):
        '''Basic Database flush'''
        self.db.insert(100, '200')
        self.db.flush()

    def test_key_count(self):
        '''Basic Database key_count'''
        self.assert_(self.db.key_count() == 0)
        for i in range(1000):
            self.db.insert(i, str(i))
            self.assert_(self.db.key_count() == i + 1)

        for i in range(1000):
            self.db.erase(i)
            self.assert_(self.db.key_count() == 1000 - (i + 1))

        self.assert_(self.db.key_count() == 0)
        for i in range(1000):
            self.db.insert(aphw.Key(i), str(i))
            self.assert_(self.db.key_count() == i + 1)

        for i in range(1000):
            self.db.erase(aphw.Key(i))
            self.assert_(self.db.key_count() == 1000 - (i + 1))

    def test_iteration(self):
        '''iteration support'''
        for i in range(1000):
            self.db.insert(i, str(i))

        found = {}
        for row in self.db:
            self.assert_(str(row.key.value) == str(row.record.data))
            found[row.key.value] = str(row.record.data)

        for i in range(100):
            self.assert_(i in found)
            self.assert_(found[i] == str(i))

    def test_roundtrip(self):
        '''populate, save, load, read'''
        
        for i in range(100):
            self.db.insert(i, str(i))
            
        self.db.close()
        
        self.db = aphw.Database(self.db_name)
        for i in range(100):
            self.assert_(str(self.db.find(i)) == str(i))

    def test_get_params(self):
        self.db.close()
        
        try:
            os.remove(self.db_name)
        except OSError:
            pass
        
        orig_params = {
            aphw.Params.CACHESIZE : 1234567,
            aphw.Params.PAGESIZE : 1024,
            aphw.Params.KEYSIZE : 23 
            }

        self.db = aphw.Database(
            self.db_name,
            params = orig_params)
                
        params = self.db.get_parameters(orig_params.keys())
        self.assert_(orig_params == params)

    def test_get_flags(self):
        expected = {
            aphw.Flags.READ_ONLY : False,
            aphw.Flags.WRITE_THROUGH : False,
            aphw.Flags.USE_BTREE : True,
            aphw.Flags.DISABLE_VAR_KEYLEN : False,
            aphw.Flags.RECORD_NUMBER : False,
            aphw.Flags.ENABLE_DUPLICATES : False,
            aphw.Flags.DISABLE_MMAP : False,
            aphw.Flags.ENABLE_RECOVERY : True, # Because enable-transactions forces it on
            aphw.Flags.ENABLE_TRANSACTIONS : True
            }
        
        flags = self.db.get_flags()
        
        for k,v in expected.items():
            self.assert_(v == bool(flags & k))

        self.db.close()
        
        try:
            os.remove(self.db_name)
        except OSError:
            pass

        self.db = aphw.Database(
            self.db_name,
            read_only=False,
            write_through=True,
            use_btree=True,
            disable_var_keylen=True,
            record_number=True,
            enable_duplicates=True,
            disable_mmap=True,
            cache_size_policy = aphw.CacheSizePolicy.NORMAL,
            enable_recovery=False,
            enable_auto_recovery=False,
            enable_transactions=False)

        expected = {
            aphw.Flags.READ_ONLY : False,
            aphw.Flags.WRITE_THROUGH : True,
            aphw.Flags.USE_BTREE : True,
            aphw.Flags.DISABLE_VAR_KEYLEN : True,
            aphw.Flags.RECORD_NUMBER : True,
            aphw.Flags.ENABLE_DUPLICATES : True,
            aphw.Flags.DISABLE_MMAP : True,
            aphw.Flags.ENABLE_RECOVERY : False, # Because enable-transactions forces it on
            aphw.Flags.ENABLE_TRANSACTIONS : False
            }

        flags = self.db.get_flags()
        
        for k,v in expected.items():
            self.assert_(v == bool(flags & k))

    def test_approx_match_type(self):
        for i in range(20):
            self.db.insert(i, 'jones')

        for i in range(20):
            k = aphw.Key(i)
            self.db.find(k)
            self.assert_(k.approximate_match_type == 0)

        for i in range(1, 20):
            k = aphw.Key(i)
            self.db.find(k, flags=aphw.Find.LT)
            self.assert_(k.value == i - 1)
            self.assert_(k.approximate_match_type == -1)

        for i in range(0, 19):
            k = aphw.Key(i)
            self.db.find(k, flags=aphw.Find.GT)
            self.assert_(k.value == i + 1)
            self.assert_(k.approximate_match_type == 1)

        self.db.insert(100, 'llama')
      
        k = aphw.Key(101)
        self.db.find(k, flags=aphw.Find.LEQ)
        self.assert_(k.value == 100)
        self.assert_(k.approximate_match_type == -1)

        k = aphw.Key(100)
        self.db.find(k, flags=aphw.Find.LEQ)
        self.assert_(k.value == 100)
        self.assert_(k.approximate_match_type == 0)

        k = aphw.Key(99)
        self.db.find(k, flags=aphw.Find.GEQ)
        self.assert_(k.value == 100)
        self.assert_(k.approximate_match_type == 1)

        k = aphw.Key(100)
        self.db.find(k, flags=aphw.Find.GEQ)
        self.assert_(k.value == 100)
        self.assert_(k.approximate_match_type == 0)

        k = aphw.Key(200)
        self.db.find(k, flags=aphw.Find.NEAR)
        self.assert_(k.value == 100)
        self.assert_(k.approximate_match_type == -1)

class ContextManagerTests(unittest.TestCase):
    def setUp(self):
        self.db_name = 'test.db'

        try:
            os.remove(self.db_name)
        except OSError:
            pass

    def tearDown(self):
        try:
            os.remove(self.db_name)
        except OSError:
            pass

    def test_basic(self):
        '''Basic context management'''
        with aphw.Database(self.db_name) as db:
            for i in range(100):
                db.insert(i, str(i))

        with aphw.Database(self.db_name) as db:
            for i in range(100):
                self.assert_(str(db.find(i)) == str(i))

    def test_exception(self):
        '''CM with exception'''

        try:
            with aphw.Database(self.db_name) as db:
                for i in range(100):
                    db.insert(i, str(i))
                raise KeyError()
        except KeyError:
            pass

        with aphw.Database(self.db_name) as db:
            for i in range(100):
                self.assert_(str(db.find(i)) == str(i))

class ReadOnlyTests(unittest.TestCase):

    def setUp(self):
        self.db_name = 'test.db'

        try:
            os.remove(self.db_name)
        except OSError:
            pass

        # We have to do a two-phase open because there is
        # no way to actually create a read-only DB from
        # scratch. You can only open a DB as read-only.

        with aphw.Database(filename=self.db_name):
            pass

        self.assert_(os.path.exists(self.db_name))

        self.db = aphw.Database(filename=self.db_name,
                                read_only=True,
                                enable_transactions=False)

    def tearDown(self):
        try:
            os.remove(self.db_name)
        except OSError:
            pass

        self.db.close()

    def test_insert(self):
        '''Read-only Database insert'''

        self.assert_(self.db.read_only)

        self.assertRaises(aphw.ReadOnlyException,
                          self.db.insert,
                          100, 'llama')

    def test_erase(self):
        '''Read-only Database erase'''

        self.assert_(self.db.read_only)

        self.assertRaises(aphw.ReadOnlyException,
                          self.db.erase,
                          100)

class RecordNumbertests(unittest.TestCase):
    
    def setUp(self):
        self.db_name = 'test.db'

        try:
            os.remove(self.db_name)
        except OSError:
            pass

        self.db = aphw.Database(self.db_name,
                                record_number=True)

    def tearDown(self):
        try:
            os.remove(self.db_name)
        except OSError:
            pass

        self.db.close()

    def test_insert(self):
        '''Insert with empty Key'''
        
        k1 = aphw.Key()
        k2 = aphw.Key()
        self.db.insert(k1, 'foo')
        self.db.insert(k2, 'bar')
        self.assert_(k1.value != k2.value)
        self.assert_(str(self.db.find(k1)) == 'foo')
        self.assert_(str(self.db.find(k2)) == 'bar')

class PartialTests(unittest.TestCase):

    def setUp(self):
        self.db_name = 'test.db'

        try:
            os.remove(self.db_name)
        except OSError:
            pass

        self.db = aphw.Database(self.db_name)

    def tearDown(self):
        self.db.close()

        try:
            os.remove(self.db_name)
        except OSError:
            pass

    def test_find(self):
        data = '1234567890' * 10
        self.db.insert(1, data)

        for b in range(len(data) + 1):
            for e in range(b + 1, len(data) + 1):
                r = self.db.find(1, 
                                 partial_offset = b,
                                 partial_size = e - b)
                
                self.assert_(str(r) == data[b:e])

        self.assertRaises(
            aphw.StatusError,
            self.db.find,
            1,
            partial_offset=1000000,
            partial_size=10)

        self.assertRaises(
            OverflowError,
            self.db.find,
            1,
            partial_offset=-1,
            partial_size=10)

    def test_insert(self):
        self.db.insert(1, 'x' * 1000)

        self.db.insert(1, 'r' * 10, partial_offset = 10, partial_size=10)
        r = self.db.find(1, partial_offset = 10, partial_size=10)
        self.assert_(str(r) == 'r' * 10)

def suite():
    cases = [Tests, 
             ContextManagerTests, 
             CtorTests, 
             PartialTests,
             ReadOnlyTests,
             ]
    suites = [unittest.TestLoader().loadTestsFromTestCase(c) for c in cases]
    return unittest.TestSuite(suites)

if __name__ == '__main__':
    unittest.TextTestRunner(verbosity=2).run(suite())
